package flink_p1

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 针对每辆车 每10s钟计算一次最近1分钟的平均速度： keyed + time + 滑动窗口
 */
object FlinkTest_16_window_agg {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)


    socketStream.map(data => {
      val srr: Array[String] = data.split(" ")
      (srr(0), srr(1).toInt) //  <=  toInt 细节需要注意！！
    }).keyBy(_._1)
      .timeWindow(Time.seconds(60), Time.seconds(10))
      .aggregate(new AggregateFunction[(String, Int), (String, Int, Int), (String, Int)] {

        override def createAccumulator(): (String, Int, Int) = ("", 0, 0)

        override def add(value: (String, Int), accumulator: (String, Int, Int)): (String, Int, Int) = {
          (value._1, accumulator._2 + 1, accumulator._3 + value._2)
        }

        override def getResult(accumulator: (String, Int, Int)): (String, Int) = {
          (accumulator._1, accumulator._3 / accumulator._2)
        }

        override def merge(a: (String, Int, Int), b: (String, Int, Int)): (String, Int, Int) = {
          (a._1, a._2 + b._2, a._3 + b._3)
        }
      }).print()

    env.execute()
  }

}
